package com.mini.task;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import com.mini.es.model.EsBusinessStudentBigLessonXcx;
import com.mini.es.model.EsBusinessStudentOrderInfo;
import com.mini.es.service.EsBusinessStudentBigLessonXcxService;
import com.mini.es.service.EsBusinessStudentOrderInfoService;
import com.mini.mysql.model.BusinessStudentBigLessonXcx;
import com.mini.mysql.model.BusinessStudentSecondOrderInfo;
import com.mini.mysql.model.Dept;
import com.mini.mysql.model.UserFullDept;
import com.mini.mysql.repository.DeptRepository;
import com.mini.mysql.repository.UserFullDeptRepository;
import com.mini.mysql.service.BusinessStudentBigLessonXcxService;
import com.mini.mysql.service.BusinessStudentSecondOrderInfoService;
import com.mini.mysql.service.MnsService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.math.BigInteger;
import java.util.*;
import java.util.stream.Collectors;

import static com.mini.task.TaskAsyncMiniOrderDataServiceImpl.df;

@Service
@Slf4j
public class TaskAsyncBigOrderDataServiceImpl {


    @Resource(name = "asyncBigOrderDataTaskExecutor")
    private ThreadPoolTaskExecutor asyncBigOrderDataTaskExecutor;

    @Resource(name = "asyncSecondOrderInfoDataTaskExecutor")
    private ThreadPoolTaskExecutor asyncSecondOrderInfoDataTaskExecutor;

    @Autowired
    private EsBusinessStudentBigLessonXcxService esBusinessStudentBigLessonXcxService;
    @Autowired
    private BusinessStudentBigLessonXcxService businessStudentBigLessonXcxService;
    @Autowired
    private EsBusinessStudentOrderInfoService esBusinessStudentOrderInfoService;
    @Autowired
    private BusinessStudentSecondOrderInfoService businessStudentSecondOrderInfoService;
    @Autowired
    private DeptRepository deptRepository;
    @Autowired
    private UserFullDeptRepository userFullDeptRepository;
    @Autowired
    private MnsService mnsService;

    public static final Date initTime = DateUtil.parse("2020-01-01", DatePattern.NORM_DATE_PATTERN);

    @Value("${spring.profiles.active}")
    private String profiles;


    /**
     * 根据updateTime 从mysql传输数据 order表数据(使用线程池 asyncDataTaskExecutor)
     */
    @Scheduled(cron = "0/30 * * * * ?")
    public void asyncBigOrderDataFromMysql() {
        if (
                profiles.contains("local") ||
                        profiles.contains("dev")) {
            return;
        }
        asyncBigOrderDataTaskExecutor.submit(() -> {
            try {
                //从mysql拉取大课订单数据，任务结束后，等待100毫秒，从ES中拉取小课订单数据，与大课数据进行关联合并(串行任务，防止出现数据覆盖)
                asyncBigOrderDataFromMysql(null);
                Thread.sleep(100);
                asyncBigOrderDataFromEs(null);
                Thread.sleep(100);
                extractedBigOrderSecondOrder(null);
                Thread.sleep(100);
                extractedBigOrderDept();
                Thread.sleep(100);
                extractedBigOrderUserFullDept(null);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

    }

//    @Scheduled(cron = "0/5 * * * * ?")
//    public void asyncSecondOrderData() {
//        if (
//                profiles.contains("local") ||
//                        profiles.contains("dev")) {
//            return;
//        }
//        asyncSecondOrderInfoDataTaskExecutor.submit(() -> {
//            extractedBigOrderSecondOrder(null);
//        });
//
//    }


    public void asyncBigOrderDataFromMysql(Date beginTime) {
        log.info("-----开始同步big lesson from mysql数据");
        //每次同步数据条数
        int pageSize = 10000;
        final Date[] orderStartTime = {beginTime};
        EsBusinessStudentBigLessonXcx order = esBusinessStudentBigLessonXcxService.findLatestUpdateTimeData();
        if (Objects.isNull(orderStartTime[0]) && Objects.nonNull(order) && Objects.nonNull(order.getBigLessUpdateTime())) {
            orderStartTime[0] = order.getBigLessUpdateTime();
        }

        if (Objects.isNull(orderStartTime[0])) {
            orderStartTime[0] = TaskAsyncMiniOrderDataServiceImpl.initTime;
        }

        Date now = new Date();
        Page<BusinessStudentBigLessonXcx> page = null;
        Integer total = businessStudentBigLessonXcxService.countOrder(orderStartTime[0], now);
        int i = 0;
        try {
            long startTime = System.currentTimeMillis();
            final int[] pageNum = {0};
            while (i == 0 || page.hasNext()) {
                Pageable pageable = PageRequest.of(pageNum[0], pageSize, Sort.Direction.ASC, "updateTime", "orderNo");
                i++;

                //mysql
                page = businessStudentBigLessonXcxService.page(orderStartTime[0], now, pageable);
                int size = Math.min(pageSize * i, total);
                List<EsBusinessStudentBigLessonXcx> newList = page.getContent().stream().map(EsBusinessStudentBigLessonXcx::from).collect(Collectors.toList());
                if (newList.isEmpty()) {
                    log.info("-----任务结束-同步数据 条数 c = {} 进度： {}%", size, df.format((float) size / total * 100));
                    return;
                }

                //ES exist data
                List<String> orderNoList = newList.stream().map(EsBusinessStudentBigLessonXcx::getOrderNo).distinct().filter(Objects::nonNull).collect(Collectors.toList());
                List<EsBusinessStudentBigLessonXcx> existList = esBusinessStudentBigLessonXcxService.findAllById(orderNoList);

                //如果updateTime字段改变，则修改updateTime请求第1页数据，如果updateTime字段不变，则页码递增，请求第二页数据（优化查询速度）
                newList.stream().filter(o -> Objects.nonNull(o.getBigLessUpdateTime())).max(Comparator.comparing(EsBusinessStudentBigLessonXcx::getBigLessUpdateTime))
                        .ifPresent(o -> {
                            if (o.getBigLessUpdateTime().compareTo(orderStartTime[0]) == 0) {
                                pageNum[0]++;
                            } else {
                                orderStartTime[0] = o.getBigLessUpdateTime();
                                pageNum[0] = 0;
                            }
                        });
                //如果是已经存在的数据，不修改attachUpdateTime字段，防止数据互相覆盖造成误差
                newList = newList.stream().filter(newOrder -> {
                    EsBusinessStudentBigLessonXcx existOrder = existList.stream().filter(exist -> newOrder.getOrderNo().equals(exist.getOrderNo())).findFirst().orElse(null);
                    //es中不存在
                    if (Objects.isNull(existOrder)) {
                        return true;
                    }
                    //es中与mysql中数据相同
                    if (equal(newOrder, existOrder)) {
                        return false;
                    }
                    //数据不同
                    newOrder.setMysqlUpdateTime(existOrder.getMysqlUpdateTime());
                    if((newOrder.getOrderState().equals("15") || newOrder.getOrderState().equals("40")) && Objects.isNull(newOrder.getPayFinishTime())){
                        newOrder.setPayFinishTime(newOrder.getOrderCreateTime());
                    }
                    return true;
                }).collect(Collectors.toList());

                if (newList.isEmpty()) {
                    log.info("-----任务结束-同步数据 条数 c = {} 进度： {}%", size, df.format((float) size / total * 100));
                    return;
                }

                Set<Long> secondOrderIds = newList.stream().filter(o -> Objects.nonNull(o.getSecondOrderId())).map(o -> o.getSecondOrderId()).collect(Collectors.toSet());
                List<BusinessStudentSecondOrderInfo> secondOrderInfos = businessStudentSecondOrderInfoService.findByIds(secondOrderIds);
                Map<Long, List<BusinessStudentSecondOrderInfo>> listMap = secondOrderInfos.stream().collect(Collectors.groupingBy(o -> o.getId()));

                //查询关联小课订单数据，进行数据合并
                List<String> miniOrderNos = newList.stream().map(EsBusinessStudentBigLessonXcx::getMiniOrderNo).filter(Objects::nonNull).distinct().collect(Collectors.toList());
                List<EsBusinessStudentOrderInfo> miniOrderList = esBusinessStudentOrderInfoService.findAllById(miniOrderNos);
                newList.forEach(bigOrder -> {

                    List<BusinessStudentSecondOrderInfo> secondOrderInfoList = listMap.get(bigOrder.getSecondOrderId());
                    if (Objects.nonNull(bigOrder.getSecondOrderId()) && CollectionUtils.isNotEmpty(secondOrderInfoList)) {
                        BusinessStudentSecondOrderInfo secondOrderInfo = secondOrderInfoList.get(0);
                        bigOrder.setChanceStates(secondOrderInfo.getChanceStates());
                        bigOrder.setContinueType(secondOrderInfo.getContinueType());
                        bigOrder.setSecondTeacherId(secondOrderInfo.getSecondTeacherId());
                        bigOrder.setSecondTeacherName(secondOrderInfo.getSecondTeacherName());
                        bigOrder.setSecondOrderUpdateTime(secondOrderInfo.getUpdateTime());
                    }

                    EsBusinessStudentOrderInfo miniOrderObj = miniOrderList.stream().filter(miniOrder -> miniOrder.getOrderNo().equals(bigOrder.getMiniOrderNo())).findFirst().orElse(null);
                    //计算未关联大课信息
                    bigOrder.unRefData();
                    if (Objects.nonNull(miniOrderObj)) {
                        bigOrder.extractMiniOrder(miniOrderObj);
                        //计算一阶大课信息
                        bigOrder.oneStageData();
                    }
                    //teacherId默认值用于统计
                    if (Objects.isNull(bigOrder.getTeacherId())) {
                        bigOrder.setTeacherId(BigInteger.ONE);
                    }

                });
                esBusinessStudentBigLessonXcxService.save(newList);

                long endTime = System.currentTimeMillis();
                float completion = (float) size / total;
                long totalSecond = (long) ((endTime - startTime) / 1000 / completion);
                log.info("本次同步数据 总条数 = {} 本次处理条数 c = {} 进度： {}% 需要总时长 {}分钟, 预计还需要{}分钟", total, size, df.format(completion * 100), df.format(totalSecond / 60), df.format((float) totalSecond / 60 * (1 - completion)));
            }
        } catch (Exception e) {
            log.error("", e);
        }

        log.info("结束同步bigLesson-From-Mysql数据，同步数据条数 {}", total);

    }


    public void asyncBigOrderDataFromEs(Date beginTime) {

        log.info("-----开始同步big lesson from es 数据");
        //每次同步数据条数
        int pageSize = 10000;
        final Date[] orderStartTime = {beginTime};
        EsBusinessStudentBigLessonXcx order = esBusinessStudentBigLessonXcxService.findLatestMysqlUpdateTimeData();
        if (Objects.isNull(orderStartTime[0]) && Objects.nonNull(order) && Objects.nonNull(order.getMysqlUpdateTime())) {
            orderStartTime[0] = order.getMysqlUpdateTime();
        }

        if (Objects.isNull(orderStartTime[0])) {
            orderStartTime[0] = TaskAsyncMiniOrderDataServiceImpl.initTime;
        }

        Date now = new Date();
        final int[] pageNum = {0};
        Page<EsBusinessStudentOrderInfo> page = null;
        Integer total = esBusinessStudentOrderInfoService.countByMysqlUpdateTimeBetween(orderStartTime[0], now);
        int i = 0;
        try {
            long startTime = System.currentTimeMillis();
            while (i == 0 || page.hasNext()) {
                Pageable pageable = PageRequest.of(pageNum[0], pageSize, Sort.Direction.ASC, "mysqlUpdateTime");
                i++;
                page = esBusinessStudentOrderInfoService.pageBetweenMysqlUpdateTime(orderStartTime[0], now, pageable);
                int size = Math.min(pageSize * i, total);
                List<EsBusinessStudentOrderInfo> newList = page.getContent();
                if (newList.isEmpty()) {
                    log.info("-----任务结束-同步数据 条数 c = {} 进度： {}%", size, df.format((float) size / total * 100));
                    return;
                }

                //如果updateTime字段改变，则修改updateTime请求第1页数据，如果updateTime字段不变，则页码递增，请求第二页数据（优化查询速度）
                newList.stream().filter(o -> Objects.nonNull(o.getMysqlUpdateTime()))
                        .max(Comparator.comparing(EsBusinessStudentOrderInfo::getMysqlUpdateTime))
                        .ifPresent(o -> {
                            if (o.getMysqlUpdateTime().compareTo(orderStartTime[0]) == 0) {
                                pageNum[0]++;
                            } else {
                                orderStartTime[0] = o.getMysqlUpdateTime();
                                pageNum[0] = 0;
                            }
                        });
                List<String> miniOrderNoList = newList.stream().map(EsBusinessStudentOrderInfo::getOrderNo).distinct().filter(Objects::nonNull).collect(Collectors.toList());
                List<EsBusinessStudentBigLessonXcx> existList = esBusinessStudentBigLessonXcxService.findByMiniOrderNoIn(miniOrderNoList);
                //大课表中不存在关联数据，继续下一段内容处理
                if (existList.isEmpty()) {
                    continue;
                }

                //如果是已经存在的数据，不修改attachUpdateTime字段，防止数据互相覆盖造成误差
                existList = existList.stream().filter(existOrder -> {
                    if (Objects.isNull(existOrder.getMiniOrderNo())) {
                        return false;
                    }
                    EsBusinessStudentOrderInfo miniOrder = newList.stream().filter(mini -> mini.getOrderNo().equals(existOrder.getMiniOrderNo())).findFirst().orElse(null);
                    if (Objects.isNull(miniOrder)) {
                        return false;
                    }

                    if (TaskAsyncMiniOrderDataServiceImpl.equal(miniOrder.getMysqlUpdateTime(), existOrder.getMysqlUpdateTime())) {
                        return false;
                    }
                    //数据合并
                    existOrder.extractMiniOrder(miniOrder);
                    return true;
                }).collect(Collectors.toList());

                //保存数据
                esBusinessStudentBigLessonXcxService.save(existList);
                long endTime = System.currentTimeMillis();
                float completion = (float) size / total;
                long totalSecond = (long) ((endTime - startTime) / 1000 / completion);
                log.info("本次同步数据 总条数 = {} 本次处理条数 c = {} 进度： {}% 需要总时长 {}分钟, 预计还需要{}分钟", total, size, df.format(completion * 100), df.format(totalSecond / 60), df.format((float) totalSecond / 60 * (1 - completion)));
            }
        } catch (Exception e) {
            log.error("", e);
        }

        log.info("结束同步BigLesson-From-Es数据，同步数据条数 {}", total);
    }


    private boolean equal(EsBusinessStudentBigLessonXcx one, EsBusinessStudentBigLessonXcx two) {
        return false;
//        return TaskAsyncMiniOrderDataServiceImpl.equal(one.getMysqlUpdateTime(), two.getMysqlUpdateTime())
//                && TaskAsyncMiniOrderDataServiceImpl.equal(one.getBigLessUpdateTime(), two.getBigLessUpdateTime());

    }


    public void extractedBigOrderDept() {
        log.info("-----开始同步dept中的pid");
        try {
            List<Dept> depts = deptRepository.findAll();
            updateByDept(depts);
        } catch (Exception e) {
            log.error("", e);
        }
    }

    private void updateByDept(List<Dept> depts) {
        if (depts.isEmpty()) {
            log.info("同步数据-depts，任务结束！");
        }
        int total = 0;
        //每次同步数据条数
        int pageSize = 10000;
        Pageable pageable = PageRequest.of(0, pageSize);
        for (int i = 0; i < depts.size(); i++) {
            Dept dept = depts.get(i);
            Page<EsBusinessStudentBigLessonXcx> existList = null;
            while (Objects.isNull(existList) || existList.hasNext()) {
                existList = esBusinessStudentBigLessonXcxService.pageBySku(dept.getId(), pageable);
                if (existList.getContent().isEmpty()) {
                    log.debug("dept {} 没有查询到需要修改的数据，进行下一个任务", dept.getId());
                    break;
                }
                existList.getContent().forEach(t -> {
                    t.setPid(dept.getPid());

                });
                esBusinessStudentBigLessonXcxService.save(existList.getContent());
                total += existList.getContent().size();
//                log.info("本次同步数据-secondOrderInfo = {}, 总条数 = {} 本次处理条数 {}, 已经修改secondOrderInfo size = {} 一共需要修改的 size = {}", secondOrderInfo.getId(), existList.getTotalElements(), existList.getContent().size(), i + 1, secondOrderInfos.size());

            }
        }
        log.info("本次同步数据-dpet结束! 同步数据 total = {}", total);
    }

    public void extractedBigOrderUserFullDept(Date beginTime) {
        log.info("-----开始同步second_order_info数据");
        final Date[] orderStartTime = {beginTime};
        EsBusinessStudentBigLessonXcx bigLessonXcx = esBusinessStudentBigLessonXcxService.findLatestDeptUpdateTimeData();
        if (Objects.isNull(orderStartTime[0]) && Objects.nonNull(bigLessonXcx) && Objects.nonNull(bigLessonXcx.getSecondOrderUpdateTime())) {
            orderStartTime[0] = bigLessonXcx.getSecondOrderUpdateTime();
        }
        if (Objects.isNull(orderStartTime[0])) {
            orderStartTime[0] = initTime;
        }

        Date now = new Date();
        try {
            List<UserFullDept> userFullDeptList = userFullDeptRepository.findByCheckTimeBetweenOrderByCheckTimeDesc(orderStartTime[0], now);
            updateByFullDept(userFullDeptList);
        } catch (Exception e) {
            log.error("", e);
        }
    }

    private void updateByFullDept(List<UserFullDept> userFullDeptList) {
        if (userFullDeptList.isEmpty()) {
            log.info("同步数据-UserFullDept为空，任务结束！");
        }
        int total = 0;
        //每次同步数据条数
        int pageSize = 10000;
        Pageable pageable = PageRequest.of(0, pageSize);
        for (int i = 0; i < userFullDeptList.size(); i++) {
            UserFullDept userFullDept = userFullDeptList.get(i);
            Page<EsBusinessStudentBigLessonXcx> existList = null;
            while (Objects.isNull(existList) || existList.hasNext()) {
                existList = esBusinessStudentBigLessonXcxService.pageByMiniTeacherId(userFullDept.getUserId(), userFullDept.getCheckTime(), pageable);
                if (existList.getContent().isEmpty()) {
                    log.debug("pageByMiniTeacherId {} 没有查询到需要修改的数据，进行下一个任务", userFullDept.getUserId());
                    break;
                }
                existList.getContent().forEach(t -> {
                    t.setU1DeptLevel3Id(userFullDept.getDeptLevel3Id());
                    t.setU1Od(userFullDept.getOd());
                    t.setU1Og(userFullDept.getOg());
                    t.setU1Sku(userFullDept.getSku());
                    t.setU1WorkRoom(userFullDept.getWorkRoom());
                    t.setDeptUpdateTime(userFullDept.getCheckTime());
                });
                esBusinessStudentBigLessonXcxService.save(existList.getContent());
                total += existList.getContent().size();
//                log.info("本次同步数据-secondOrderInfo = {}, 总条数 = {} 本次处理条数 {}, 已经修改secondOrderInfo size = {} 一共需要修改的 size = {}", secondOrderInfo.getId(), existList.getTotalElements(), existList.getContent().size(), i + 1, secondOrderInfos.size());

            }
        }

        for (int i = 0; i < userFullDeptList.size(); i++) {
            UserFullDept userFullDept = userFullDeptList.get(i);
            Page<EsBusinessStudentBigLessonXcx> existList = null;
            while (Objects.isNull(existList) || existList.hasNext()) {
                existList = esBusinessStudentBigLessonXcxService.pageBySecondTeacherId(userFullDept.getUserId(), userFullDept.getCheckTime(), pageable);
                if (existList.getContent().isEmpty()) {
                    log.debug("pageBySecondOrderId {} 没有查询到需要修改的数据，进行下一个任务", userFullDept.getUserId());
                    break;
                }
                existList.getContent().forEach(t -> {
                    t.setU2DeptLevel3Id(userFullDept.getDeptLevel3Id());
                    t.setU2Od(userFullDept.getOd());
                    t.setU2Og(userFullDept.getOg());
                    t.setU2Sku(userFullDept.getSku());
                    t.setU2WorkRoom(userFullDept.getWorkRoom());
                    t.setDeptUpdateTime(userFullDept.getCheckTime());
                });
                esBusinessStudentBigLessonXcxService.save(existList.getContent());
                total += existList.getContent().size();
//                log.info("本次同步数据-secondOrderInfo = {}, 总条数 = {} 本次处理条数 {}, 已经修改secondOrderInfo size = {} 一共需要修改的 size = {}", secondOrderInfo.getId(), existList.getTotalElements(), existList.getContent().size(), i + 1, secondOrderInfos.size());

            }
        }
        log.info("本次同步数据-UserFullDept结束! 同步数据 total = {}", total);
    }

    public void extractedBigOrderSecondOrder(Date beginTime) {
        log.info("-----开始同步second_order_info数据");
        final Date[] orderStartTime = {beginTime};
        EsBusinessStudentBigLessonXcx bigLessonXcx = esBusinessStudentBigLessonXcxService.findLatestSecondOrderUpdateTimeData();
        if (Objects.isNull(orderStartTime[0]) && Objects.nonNull(bigLessonXcx) && Objects.nonNull(bigLessonXcx.getSecondOrderUpdateTime())) {
            orderStartTime[0] = bigLessonXcx.getSecondOrderUpdateTime();
        }
        if (Objects.isNull(orderStartTime[0])) {
            orderStartTime[0] = initTime;
        }

        Date now = new Date();
        try {
            List<BusinessStudentSecondOrderInfo> secondOrderInfos = businessStudentSecondOrderInfoService.findByUpdateTimeBetween(orderStartTime[0], now);
            updateBySecondOrder(secondOrderInfos);
        } catch (Exception e) {
            log.error("", e);
        }
    }

    private void updateBySecondOrder(List<BusinessStudentSecondOrderInfo> secondOrderInfos) {
        if (secondOrderInfos.isEmpty()) {
            log.info("同步数据-SecondOrderInfo为空，任务结束！");
        }
        int total = 0;
        //每次同步数据条数
        int pageSize = 10000;
        Pageable pageable = PageRequest.of(0, pageSize);
        for (int i = 0; i < secondOrderInfos.size(); i++) {
            BusinessStudentSecondOrderInfo secondOrderInfo = secondOrderInfos.get(i);
            if (secondOrderInfo.getId() == 19092) {
                log.info("{}", secondOrderInfo);
            }
            Page<EsBusinessStudentBigLessonXcx> existList = null;
            while (Objects.isNull(existList) || existList.hasNext()) {
                existList = esBusinessStudentBigLessonXcxService.pageBySecondOrderId(secondOrderInfo.getId(), secondOrderInfo.getUpdateTime(), pageable);
                if (existList.getContent().isEmpty()) {
                    log.debug("secondOrderInfo {} 没有查询到需要修改的数据，进行下一个任务", secondOrderInfo.getId());
                    break;
                }
                existList.getContent().forEach(t -> {
                    t.setChanceStates(secondOrderInfo.getChanceStates());
                    t.setContinueType(secondOrderInfo.getContinueType());
                    t.setSecondTeacherId(secondOrderInfo.getSecondTeacherId());
                    t.setSecondTeacherName(secondOrderInfo.getSecondTeacherName());
                    t.setSecondOrderUpdateTime(secondOrderInfo.getUpdateTime());
                });
                esBusinessStudentBigLessonXcxService.save(existList.getContent());
                total += existList.getContent().size();
//                log.info("本次同步数据-secondOrderInfo = {}, 总条数 = {} 本次处理条数 {}, 已经修改secondOrderInfo size = {} 一共需要修改的 size = {}", secondOrderInfo.getId(), existList.getTotalElements(), existList.getContent().size(), i + 1, secondOrderInfos.size());

            }
        }
        log.info("本次同步数据-secondOrderInfo结束! 同步数据 total = {}", total);
    }


}